package com.shengzai.flink.state

import com.alibaba.fastjson2.{JSON, JSONObject}
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.{lang, util}



object Demo2ListState {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val socketDs: DataStream[String] = env.socketTextStream("master", 8888)

    //解析数据
    val mapDS: DataStream[(Long, Double)] = socketDs.map(
      line => {
        val jsonObj: JSONObject = JSON.parseObject(line)
        val road: Long = jsonObj.getLong("34308116")
        val speed: Double = jsonObj.getDouble("speed")
        (road, speed)
      }
    )
    //初始化listState
    var speeds :ListState[Double] =_
    //分组key
    val keyedS: KeyedStream[(Long, Double), Long] = mapDS.keyBy(_._1)

    keyedS.process(new KeyedProcessFunction[Long,(Long,Double),(Long,Double)] {


      override def open(parameters: Configuration): Unit = {
        //获取RuntimeContext对象
        val context: RuntimeContext = getRuntimeContext
        //创建listState
        val listState = new ListStateDescriptor[Double]("spends", classOf[Double])
        //初始化listState
        speeds =context.getListState(listState)
      }

      override def processElement(value: (Long, Double),
                                  ctx: KeyedProcessFunction[Long, (Long, Double),
                                    (Long, Double)]#Context, out: Collector[(Long, Double)]): Unit = {
        val (road:Long,speed:Double) = value
        speeds.add(speed)
        val int: util.Iterator[Double] = speeds.get().iterator()
        int.next()

      }
    })




  }

}
